package t20240724;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Timestamp;


/**
 last_value：
   返回最后一个非null值。

 */
public class D3_rownumber {


    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        Table table = tableEnv.fromValues(
                DataTypes.ROW(
                        DataTypes.FIELD("stime", DataTypes.BIGINT()),
                        DataTypes.FIELD("name", DataTypes.STRING()),
                        DataTypes.FIELD("val", DataTypes.INT())
                ),

                Row.of(1, "aa", 1),
                Row.of(1, "bb", 1),
                Row.of(1, "cc", 1),

                Row.of(2, "aa", 2),
                Row.of(2, "bb", 2),
                Row.of(2, "cc", 2),

                Row.of(3, "aa", 3),
                Row.of(3, "bb", 3),
                Row.of(3, "cc", 3)
        );

        tableEnv.registerTable("ods_tb", table);


        String sink = "CREATE TABLE pprint (" +"    " +
                "    stime BIGINT, " +
                "    name STRING, " +
                "    val INT, " +
                "    rownum BIGINT " +
                ") WITH (" +
                "    'connector' = 'print'" +
                ")";

        String sql = "insert into pprint " +
                " SELECT stime,name,val,rownum FROM " +
                " (SELECT *,ROW_NUMBER() OVER(PARTITION BY name ORDER BY stime DESC) as rownum FROM ods_tb) " +
                " WHERE rownum = 1";
        System.out.println(sql);
        tableEnv.executeSql(sink);
        tableEnv.executeSql(sql);

        System.out.println(new Timestamp(System.currentTimeMillis()));



    }
}
